package com.shujia.spark.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo11Work {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("hive")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._

    //读取数据
    val wordsDF: DataFrame = spark.read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING,burk_name STRING,sTime STRING")
      .load("data/works.txt")

    wordsDF
      //取出年月日
      .withColumn("sDate", date_format($"sTime", "yyyy-MM-dd"))
      //获取上一个月工作的公司
      .withColumn("last_burk_name", lag($"burk_name", 1) over Window.partitionBy($"id").orderBy($"sDate"))
      //打上标记
      .withColumn("flag", when($"burk_name" === $"last_burk_name", 1).otherwise(1))
      //打上类别编号
      .withColumn("clazz", sum($"flag") over Window.partitionBy($"id").orderBy($"sDate"))
      .groupBy($"id", $"burk_name", $"clazz")
      .agg(
        min($"sDate") as "start_date",
        max($"sDate") as "end_date"
      )
    // .show()


    wordsDF.createOrReplaceTempView("words")

    spark.sql(
      """
        |select
        |    id,
        |    burk_name,
        |    clazz,
        |    min(sDate) as start_date,
        |    max(sDate) as end_date
        |from (
        |    select
        |        id,
        |        burk_name,
        |        sDate,
        |        flag,
        |        sum(flag) over(partition by id order by sDate) as clazz -- 增加类别
        |    from (
        |        select
        |            id,
        |            burk_name,
        |            sDate,
        |            case when burk_name=last_burk_name then 0 else 1 end as flag -- 打上标记
        |        from (
        |            select
        |                id,
        |                burk_name,
        |                date_format(sTime,'yyyy-MM-dd') as sDate,
        |                lag(burk_name,1) over(partition by id order by date_format(sTime,'yyyy-MM-dd')) as last_burk_name -- 取出上一家工作的公司
        |            from
        |                words
        |        )
        |    )
        |)
        |group by
        |    id,
        |    burk_name,
        |    clazz
        |""".stripMargin).show()

  }


}
